package org.budo.dubbo.protocol.dubbo;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
import com.alibaba.dubbo.common.utils.ConfigUtils;
import com.alibaba.dubbo.common.utils.NetUtils;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.remoting.Channel;
import com.alibaba.dubbo.remoting.RemotingException;
import com.alibaba.dubbo.remoting.Transporter;
import com.alibaba.dubbo.remoting.exchange.ExchangeChannel;
import com.alibaba.dubbo.remoting.exchange.ExchangeClient;
import com.alibaba.dubbo.remoting.exchange.ExchangeHandler;
import com.alibaba.dubbo.remoting.exchange.ExchangeServer;
import com.alibaba.dubbo.remoting.exchange.Exchangers;
import com.alibaba.dubbo.remoting.exchange.support.ExchangeHandlerAdapter;
import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcContext;
import com.alibaba.dubbo.rpc.RpcException;
import com.alibaba.dubbo.rpc.RpcInvocation;
import com.alibaba.dubbo.rpc.protocol.dubbo.DubboCodec;
import com.alibaba.dubbo.rpc.protocol.dubbo.DubboExporter;
import com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker;
import com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol;
import com.alibaba.dubbo.rpc.support.ProtocolUtils;

/**
 * 抽象父类
 * 
 * @author lmw
 */
public class AbstractBudoDubboProtocol extends DubboProtocol {
    public static final String COMPATIBLE_CODEC_NAME = "dubbo1compatible";

    public static final int DEFAULT_PORT = 20880;

    public final ReentrantLock lock = new ReentrantLock();

    private final Map<String, ExchangeServer> serverMap = new ConcurrentHashMap<String, ExchangeServer>(); // <host:port,Exchanger>

    private final Map<String, BudoReferenceCountExchangeClient> referenceClientMap = new ConcurrentHashMap<String, BudoReferenceCountExchangeClient>(); // <host:port,Exchanger>

    private final ConcurrentMap<String, BudoLazyConnectExchangeClient> ghostClientMap = new ConcurrentHashMap<String, BudoLazyConnectExchangeClient>();

    // consumer side export a stub service for dispatching event
    // servicekey-stubmethods
    private final ConcurrentMap<String, String> stubServiceMethodsMap = new ConcurrentHashMap<String, String>();

    protected static final String IS_CALLBACK_SERVICE_INVOKE = "_isCallBackServiceInvoke";

    private ExchangeHandler requestHandler;

    public ExchangeHandler getRequestHandler() {
        if (null == requestHandler) {
            this.initRequestHandler();
        }
        return requestHandler;
    }

    private synchronized void initRequestHandler() {
        if (null == requestHandler) {
            requestHandler = new ExchangeHandlerAdapter() {
                public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
                    if (message instanceof Invocation) {
                        Invocation invocation = (Invocation) message;
                        Invoker<?> invoker = AbstractBudoDubboProtocol.this.getInvoker(channel, invocation);
                        // 如果是callback 需要处理高版本调用低版本的问题
                        if (Boolean.TRUE.toString().equals(invocation.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                            String methodsStr = invoker.getUrl().getParameters().get("methods");
                            boolean hasMethod = false;
                            if (methodsStr == null || methodsStr.indexOf(",") == -1) {
                                hasMethod = invocation.getMethodName().equals(methodsStr);
                            } else {
                                String[] methods = methodsStr.split(",");
                                for (String method : methods) {
                                    if (invocation.getMethodName().equals(method)) {
                                        hasMethod = true;
                                        break;
                                    }
                                }
                            }
                            if (!hasMethod) {
                                logger.warn(new IllegalStateException("The methodName " + invocation.getMethodName() + " not found in callback service interface ,invoke will be ignored. please update the api interface. url is:" + invoker.getUrl()) + " ,invocation is :" + invocation);
                                return null;
                            }
                        }
                        RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                        return invoker.invoke(invocation);
                    }

                    throw new RemotingException(channel, "Unsupported request: " + message.getClass().getName() + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
                }

                public void received(Channel channel, Object message) throws RemotingException {
                    if (message instanceof Invocation) {
                        reply((ExchangeChannel) channel, message);
                    } else {
                        super.received(channel, message);
                    }
                }

                public void connected(Channel channel) throws RemotingException {
                    invoke(channel, Constants.ON_CONNECT_KEY);
                }

                public void disconnected(Channel channel) throws RemotingException {
                    if (logger.isInfoEnabled()) {
                        logger.info("disconected from " + channel.getRemoteAddress() + ",url:" + channel.getUrl());
                    }
                    invoke(channel, Constants.ON_DISCONNECT_KEY);
                }

                private void invoke(Channel channel, String methodKey) {
                    Invocation invocation = createInvocation(channel, channel.getUrl(), methodKey);
                    if (invocation != null) {
                        try {
                            received(channel, invocation);
                        } catch (Throwable t) {
                            logger.warn("Failed to invoke event method " + invocation.getMethodName() + "(), cause: " + t.getMessage(), t);
                        }
                    }
                }

                private Invocation createInvocation(Channel channel, URL url, String methodKey) {
                    String method = url.getParameter(methodKey);
                    if (method == null || method.length() == 0) {
                        return null;
                    }
                    RpcInvocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
                    invocation.setAttachment(Constants.PATH_KEY, url.getPath());
                    invocation.setAttachment(Constants.GROUP_KEY, url.getParameter(Constants.GROUP_KEY));
                    invocation.setAttachment(Constants.INTERFACE_KEY, url.getParameter(Constants.INTERFACE_KEY));
                    invocation.setAttachment(Constants.VERSION_KEY, url.getParameter(Constants.VERSION_KEY));
                    if (url.getParameter(Constants.STUB_EVENT_KEY, false)) {
                        invocation.setAttachment(Constants.STUB_EVENT_KEY, Boolean.TRUE.toString());
                    }
                    return invocation;
                }
            };
        }
    }

    public Collection<ExchangeServer> getServers() {
        return Collections.unmodifiableCollection(serverMap.values());
    }

    public Collection<Exporter<?>> getExporters() {
        return Collections.unmodifiableCollection(exporterMap.values());
    }

    protected Map<String, Exporter<?>> _getExporterMap() {
        return exporterMap;
    }

    protected boolean isClientSide(Channel channel) {
        InetSocketAddress address = channel.getRemoteAddress();
        URL url = channel.getUrl();
        return url.getPort() == address.getPort() && NetUtils.filterLocalHost(channel.getUrl().getIp()).equals(NetUtils.filterLocalHost(address.getAddress().getHostAddress()));
    }

    /**
     * 改进实现版本, 加了空值判断 The method AbstractDubboProtocol.getInvoker(Channel, Invocation) does not override the inherited method from DubboProtocol since it is private to a different package
     * 
     * @see com.alibaba.dubbo.rpc.protocol.dubbo.CallbackServiceCodec#decodeInvocationArgument(Channel, RpcInvocation, Class<?>[], int, Object)
     */
    @SuppressWarnings({ "all" })
    protected Invoker<?> getInvoker(Channel channel, Invocation invocation) throws RemotingException {
        return this.doGetInvokerNew(channel, invocation);
    }

    public Invoker<?> doGetInvokerNew(Channel channel, Invocation invocation) throws RemotingException {
        boolean isCallBackServiceInvoke = false;
        boolean isStubServiceInvoke = false;
        int port = channel.getLocalAddress().getPort();
        String path = invocation.getAttachments().get(Constants.PATH_KEY);

        // 如果是客户端的回调服务.
        isStubServiceInvoke = Boolean.TRUE.toString().equals(invocation.getAttachments().get(Constants.STUB_EVENT_KEY));
        if (isStubServiceInvoke) {
            port = channel.getRemoteAddress().getPort();
        }

        // callback
        isCallBackServiceInvoke = this.isClientSide(channel) && !isStubServiceInvoke;
        if (isCallBackServiceInvoke) {
            String callbackServiceKey = invocation.getAttachments().get(Constants.CALLBACK_SERVICE_KEY);
            if (null == callbackServiceKey || "null".equals(callbackServiceKey)) { // 如果取到null就不拼上
                logger.error("#201 getInvoker, callbackServiceKey=" + callbackServiceKey + ", invocation.attachments=" + invocation.getAttachments());
                // path = inv.getAttachments().get(Constants.PATH_KEY);
            } else {
                path = invocation.getAttachments().get(Constants.PATH_KEY) + "." + callbackServiceKey;
            }

            invocation.getAttachments().put(IS_CALLBACK_SERVICE_INVOKE, Boolean.TRUE.toString());
        }

        String serviceVersion = invocation.getAttachments().get(Constants.VERSION_KEY);
        String serviceGroup = invocation.getAttachments().get(Constants.GROUP_KEY);
        String serviceKey = ProtocolUtils.serviceKey(port, path, serviceVersion, serviceGroup);

        DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
        if (exporter == null) {
            throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + invocation);
        }

        return exporter.getInvoker();
    }

    public Collection<Invoker<?>> getInvokers() {
        return Collections.unmodifiableCollection(invokers);
    }

    public int getDefaultPort() {
        return DEFAULT_PORT;
    }

    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();

        // export service.
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);

        // export an stub service for dispaching event
        Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT);
        Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
        if (isStubSupportEvent && !isCallbackservice) {
            String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
            if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
                if (logger.isWarnEnabled()) {
                    logger.warn(new IllegalStateException("consumer [" + url.getParameter(Constants.INTERFACE_KEY) + "], has set stubproxy support event ,but no stub methods founded."));
                }
            } else {
                stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
            }
        }

        this.openServer(url);

        return exporter;
    }

    private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        // client 也可以暴露一个只有server可以调用的服务。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                serverMap.put(key, createServer(url));
            } else {
                // server支持reset,配合override功能使用
                server.reset(url);
            }
        }
    }

    private ExchangeServer createServer(URL url) {
        // 默认开启server关闭时发送readonly事件
        url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
        // 默认开启heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
        String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);

        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);

        url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
        ExchangeServer server;
        try {
            server = Exchangers.bind(url, this.getRequestHandler());
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        str = url.getParameter(Constants.CLIENT_KEY);
        if (str != null && str.length() > 0) {
            Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
            if (!supportedTypes.contains(str)) {
                throw new RpcException("Unsupported client type: " + str);
            }
        }
        return server;
    }

    public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        // create rpc invoker.
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);
        return invoker;
    }

    private ExchangeClient[] getClients(URL url) {
        // 是否共享连接
        boolean service_share_connect = false;
        int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
        // 如果connections不配置，则共享连接，否则每服务每连接
        if (connections == 0) {
            service_share_connect = true;
            connections = 1;
        }

        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) {
            if (service_share_connect) {
                clients[i] = getSharedClient(url);
            } else {
                clients[i] = initClient(url);
            }
        }
        return clients;
    }

    /**
     * 获取共享连接
     */
    private ExchangeClient getSharedClient(URL url) {
        String key = url.getAddress();
        BudoReferenceCountExchangeClient client = referenceClientMap.get(key);
        if (client != null) {
            if (!client.isClosed()) {
                client.incrementAndGetCount();
                return client;
            }
            // logger.warn(new IllegalStateException("client is closed,but stay in clientmap .client :"+ client));
            referenceClientMap.remove(key);
        }
        ExchangeClient exchagneclient = initClient(url);

        client = new BudoReferenceCountExchangeClient(exchagneclient, ghostClientMap);
        referenceClientMap.put(key, client);
        ghostClientMap.remove(key);
        return client;
    }

    /**
     * 创建新连接.
     */
    private ExchangeClient initClient(URL url) {

        // client type setting.
        String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));

        url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
        // 默认开启heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));

        // BIO存在严重性能问题，暂时不允许使用
        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
            throw new RpcException("Unsupported client type: " + str + "," + " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
        }

        ExchangeClient client;
        try {
            // 设置连接应该是lazy的
            if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
                client = new BudoLazyConnectExchangeClient(url, this.getRequestHandler());
            } else {
                client = Exchangers.connect(url, this.getRequestHandler());
            }
        } catch (RemotingException e) {
            throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
        }
        return client;
    }

    public void destroy() {
        for (String key : new ArrayList<String>(serverMap.keySet())) {
            ExchangeServer server = serverMap.remove(key);
            if (server != null) {
                try {
                    if (logger.isInfoEnabled()) {
                        logger.info("Close dubbo server: " + server.getLocalAddress());
                    }
                    server.close(ConfigUtils.getServerShutdownTimeout());
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        }

        for (String key : new ArrayList<String>(referenceClientMap.keySet())) {
            ExchangeClient client = referenceClientMap.remove(key);
            if (client != null) {
                try {
                    if (logger.isInfoEnabled()) {
                        logger.info("Close dubbo connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress());
                    }
                    client.close();
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        }

        for (String key : new ArrayList<String>(ghostClientMap.keySet())) {
            ExchangeClient client = ghostClientMap.remove(key);
            if (client != null) {
                try {
                    if (logger.isInfoEnabled()) {
                        logger.info("Close dubbo connect: " + client.getLocalAddress() + "-->" + client.getRemoteAddress());
                    }
                    client.close();
                } catch (Throwable t) {
                    logger.warn(t.getMessage(), t);
                }
            }
        }
        stubServiceMethodsMap.clear();
        super.destroy();
    }
}